/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.jr.wx.audio.dowloadserver.handler;

import cn.ac.iie.di.ban.jr.wx.audio.dowloadserver.common.RuntimeEnv;
import com.alibaba.fastjson.JSONObject;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class AudioDownloadThread {

    private ArrayBlockingQueue<String> jsonBlockingQueue;
    private String jobid;
    private String responseDir;
    private int timeout;
    static Logger logger = Logger.getLogger(AudioDownloadThread.class.getName());

    public AudioDownloadThread(ArrayBlockingQueue<String> jsonBlockingQueue, String jobid, String responseDir, int timeout) {
        this.jsonBlockingQueue = jsonBlockingQueue;
        this.jobid = jobid;
        this.responseDir = responseDir;
        this.timeout = timeout;
    }

    public void execute() {
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(Integer.parseInt((String) RuntimeEnv.getParam(RuntimeEnv.LIMITINT)),
                Integer.parseInt((String) RuntimeEnv.getParam(RuntimeEnv.JETTY_THREADNUM)),
                timeout,
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(Integer.parseInt((String) RuntimeEnv.getParam(RuntimeEnv.JETTY_THREADNUM))),
                new ThreadPoolExecutor.DiscardOldestPolicy());
//        String responseDir = (String) RuntimeEnv.getParam(RuntimeEnv.responseDir);
        for (String json : jsonBlockingQueue) {
            try {
                JSONObject jsonObj = JSONObject.parseObject(json);
                threadPool.execute(new AudioDownloadExe(responseDir, jsonObj));
            } catch (Exception e) {
                logger.error(jobid + "can not parse " + json, e);
            }
        }

        long begin = System.currentTimeMillis();
        while (true) {//直至所有线程结束，才继续
            int activeThread = threadPool.getActiveCount();
            if (activeThread == 0) {
                logger.info(jobid + " threadPool.getActiveCount()==0 break");
                break;
            } else {
                logger.info(jobid + " threadPool.getActiveCount()!=0 " + activeThread);
                long end = System.currentTimeMillis();
                long middle = end - begin;
                //设置超时为150ms
                if (middle >= timeout) {
                    logger.info(jobid + " timeout  " + middle);
                    break;
                } else {
                    logger.debug(jobid + " not timeout  " + middle);
                }
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    logger.error(ex.getMessage(), ex);
                }
            }
        }
        threadPool.shutdown();
    }
}
